package com.itcast.mq.bootmsg;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @Author Administrator
 * @Date 2021/11/21:12:31
 * @Description:
 */
@Slf4j
@RestController
public class MQProducer {

    @Setter
    @Getter
    static class People {
        String name;
        int age;

        public People(String name, int age) {
            this.name = name;
            this.age = age;
        }
    }


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/kafka/send")
    public String sendMessage1() throws ExecutionException, InterruptedException {
        People people;
        for (int i = 0; i < 3000000; i++) {
            people = new People("key:" + i, i);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("topic1", "key:" + i, JSON.toJSONString(people));
            log.info("success: " + i);
        }
        return "success";
    }

    @KafkaListener(topics = {"topic1"})
    public void onMessage1(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) throws InterruptedException {
        // 消费的哪个topic、partition的消息,打印出消息内容
        // System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        TimeUnit.MILLISECONDS.sleep(600);
        System.out.println("******************: size: " + records.size() + "; ossfet:" + records.get(0).offset());
        ack.acknowledge();
    }

    @GetMapping("/kafka/send1")
    public String sendMessage2() throws ExecutionException, InterruptedException {
        String people = JSON.toJSONString(new People("key:2", 30));

        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.
                send("topic3", "key:1", people);
        return "success";
    }

    @KafkaListener(topics = {"topic2"})
    public void onMessage2(){
        // 消费的哪个topic、partition的消息,打印出消息内容
        // System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());

    }
}
